package my.thesis.bolts.aggregators;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.ThroughputMetric;
import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SentimentCounter extends BaseRichBolt 
{
	private static final long serialVersionUID = 1L;
	private ConcurrentHashMap<String, Integer> countsTotal;
	private int tiToEmit;
	private OutputCollector collector;
	
	private Boolean receivedSinceLastEmit;	
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;
	
	public SentimentCounter(int tiSec)
	{
		this.tiToEmit=tiSec;
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{	
		countsTotal = new ConcurrentHashMap<String, Integer>();	
		this.collector=collector;
		
		this.receivedSinceLastEmit=false;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("sentimentCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}
		
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("countsTotal"));		
	}
	 
	 private Integer countForCountry(String country) 
	 {
		 Integer count = countsTotal.get(country);
		 return count == null ? 0 : count;	
	 }
	 
	 @Override
	 public Map<String, Object> getComponentConfiguration() 
	 {
	    Config conf = new Config();
	    conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, this.tiToEmit);
	    return conf;
	 }
	 
     private boolean isTickTuple(Tuple tuple)
	 {
		 String sourceComponent = tuple.getSourceComponent();
		 String sourceStreamId = tuple.getSourceStreamId();
		 return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) 
					 && sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);
	 }

	public void execute(Tuple tuple) 
	{
		if(isTickTuple(tuple))
		{
			if(this.receivedSinceLastEmit)
			{
				//TEST
				System.out.println("Emitting scores...");
				
				emitSentiments();
			}
			
			this.receivedSinceLastEmit=false;
		}
		else
		{
			this.receivedSinceLastEmit=true;
			
			if(metricsOn)
			{
				_throughputMetric.incrTuples();
			}
			
			String c = (String) tuple.getValueByField("country");
			int sc =  tuple.getIntegerByField("score");
			
			countsTotal.put(c, countForCountry(c) + sc);			
		}  	
	}

	private void emitSentiments()
	{
		ConcurrentHashMap<String,Integer> scores = new ConcurrentHashMap<String,Integer>(countsTotal);
		
		collector.emit(new Values(scores));
		
		countsTotal.clear();
	}
}
